package org.spark.core.transformations.java;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.spark.sparkUtil.SparkJavaContextUtil;

import java.util.Arrays;
import java.util.List;

/**
 * @description: map算子案例：将集合中每一个元素都乘以2
 */
public class Map {
    public static void main(String[] args) {
        JavaSparkContext sc = SparkJavaContextUtil.getSparkStart("map");
        // 构造集合
        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
        // 并行化集合，创建初始RDD
        JavaRDD<Integer> numberRDD = sc.parallelize(numbers);
        // 使用map算子，将集合中的每个元素都乘以2
        // map算子，是对任何类型的RDD，都可以调用的
        // 在java中，map算子接收的参数是Function对象
        // 创建的Function对象，一定会让你设置第二个泛型参数，这个泛型类型，就是返回的新元素的类型
        // 同时call()方法的返回类型，也必须与第二个泛型类型同步
        // 在call()方法内部，就可以对原始RDD中的每一个元素进行各种处理和计算，并返回一个新的元素
        // 所有新的元素就会组成一个新的RDD
        JavaRDD<Integer> multipleNumberRDD = numberRDD.map(
                new Function<Integer, Integer>() {
                    private static final long serialVersionUID = 1L;

                    // 传入call()方法的，就是1,2,3,4,5
                    // 返回的就是2,4,6,8,10
                    public Integer call(Integer v1) throws Exception {
                        return v1 * 2;
                    }
                });
        // 打印新的RDD
        multipleNumberRDD.foreach(new VoidFunction<Integer>() {
            public void call(Integer t) throws Exception {
                System.out.println(t);
            }
        });

        // 关闭JavaSparkContext
        sc.close();
    }
}
